package com.atguigu.stream.transale.withstate

import com.atguigu.stream.util.MyApp
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * description ：对获取的结果直接进行窗口，所以就不需要再使用窗口了
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/1/15 20:59
 * modified By ：
 * version:    : 1.0
 */
object Windows extends MyApp {

  override def readAndTranform(ssc: StreamingContext): Unit = {

    // 获取数据
    val source: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop101", 9999)


    // TODO source.window --> source.xx.xxx.reduceByKeyAndWindow
    // TODO source.window是对原数据直接加了窗口，后面所有转换算子都生效
    // TODO source.x1.x2.reduceByKeyAndWindow 这个只是对x2返回的结果加了窗口，窗口对于前面（source.x1）的转换算子是不生产的
    // TODO 直接对原数据加了窗口，是不能优化重复计算的值了
    source.window(Seconds(9)).transform {
      rdd =>
        rdd.flatMap(_.split("\\s+")).map((_, 1))
    }
      .reduceByKey(_ + _)
      .print()
  }

}
